package climspace.hadoop;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

@SuppressWarnings("deprecation")
public class GetHistoricalDataByStation {
	private String hostName;
	private String hdfsPort;
	private String mapredPort;
	
	public GetHistoricalDataByStation(String hostName, String hdfsPort, String mapredPort) {
		this.hostName = hostName;
		this.hdfsPort = hdfsPort;
		this.mapredPort = mapredPort;
	}
	
	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		String stationName = "";
		public void configure(JobConf conf) {
			this.stationName = conf.get("ClimSpace.Filter.StationName");
		}
		
		public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// map will filter on station
			String line = value.toString();
			
			try {
				String station = (line.substring(0,7)).trim();

				if (station.equals(stationName)) {
					output.collect(value, new Text());
				}
			} catch (NumberFormatException e) {
				System.out.println("Skipping non data lines");
			} catch (StringIndexOutOfBoundsException e) {
				System.out.println("Skipping non data lines");
			}
		}
	}

	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {	
		public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output,
				Reporter reporter) throws IOException {
			output.collect(key, null);
		}
	}
	
	/**
	 * @param args[0] input folder on HDFS
	 * @param args[1] output folder on HDFS
	 * @param args[2] station id
	 * @throws Exception
	 */
	public void run(String inputPath, String outputPath, String stationName) throws Exception {
		JobConf conf = new JobConf(GetHistoricalDataByStation.class);
		conf.setJobName("GetHistDataByStation");

//		conf.set("fs.default.name", "hdfs://localhost:9000/");
//		conf.set("mapred.job.tracker", "localhost:9001");

		conf.set("fs.default.name", "hdfs://" + hostName + ":" + hdfsPort + "/");
		conf.set("mapred.job.tracker", hostName + ":" + mapredPort);
		
		conf.set("ClimSpace.Filter.StationName", stationName);

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapperClass(Map.class);
//		conf.setCombinerClass(Reduce.class);
		conf.setReducerClass(Reduce.class);

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		Path output = new Path(outputPath);
		FileSystem fs = FileSystem.get(conf);
		fs.delete(output, true);
		
		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, output);
		
		//JobClient.runJob(conf);
		JobClient job = new JobClient(conf);
		job.submitJob(conf);
	}

	public static void main(String[] args) throws Exception {
		if (args.length != 3)
			throw new RuntimeException("Incorrent number of parameter");
		GetHistoricalDataByStation  wc = new GetHistoricalDataByStation("localhost", "9000", "9001");
		wc.run(args[0], args[1], args[2]);
	}
}
